package com.css.resoft.p2pmonitor.util;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.THBaseService;
import org.apache.hadoop.hbase.thrift2.generated.TIOError;
import org.apache.hadoop.hbase.thrift2.generated.TPut;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.xml.sax.Attributes;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;

public class SAXParserHandle extends DefaultHandler{
	
	private  static boolean isQuoteRecord = false;
	private  static boolean isQuoteHead = false;
	private  String currentTag = null;
	private  boolean isFileValid = true;
	
	private static String datadate = null;
	private static String orgname = null;
	private static String orgcode = null;
	private static String reporter = null;
	private static String checker = null;
	private static String table = null;
	
	private TTransport transport = null;
	private THBaseService.Iface client = null;
	
	private String tableLoan = "J0901";
	private String tableDebtor = "J0902";
	private String tableInvest = "J0903";
	private String family = "F1";
	
	List<TPut> putLoan = null;
	List<TPut> putDebtor = null;
	List<TPut> putInvest = null;
	
	private static int indexLoan = 0;
	private static int indexDebtor = 0;
	private static int indexInvest = 0;
	
	private Map<String,String> record = new HashMap<String,String>();;
	
	public SAXParserHandle(String host,int port){
		transport = new TSocket(host,port);
		TProtocol protocol = new TBinaryProtocol(transport);
		client = new THBaseService.Client(protocol);
	}
	
	@Override
	public void startDocument() throws SAXException{
		super.startDocument();
		putLoan = new ArrayList<TPut>();
		putDebtor = new ArrayList<TPut>();
		putInvest = new ArrayList<TPut>();
		isFileValid = true;
	}

	@Override
	public void endDocument() throws SAXException{
		super.endDocument();
	}
	
	@Override
	public void startElement(String arg0,String arg1,String name,Attributes attribtes) throws SAXException{
		super.startElement(arg0, arg1, name, attribtes);
		
		currentTag = name;
        switch(name){
        	case "record":isQuoteRecord = true;break;
        	case "tableP":isQuoteHead = true;break;
        	case "report1":table = tableLoan;break;
        	case "report2":table = tableDebtor;break;
        	case "report3":table = tableInvest;break;
        }	
	}
	
    @Override
    public void endElement(String arg0, String arg1, String name) throws SAXException {
        super.endElement(arg0, arg1, name);
        
        if(name.equals("tableP")) 
        	isQuoteHead = false;
    	
        if(name.equals("record")){
        	addData();
        	record.clear();
        	isQuoteRecord = false;
        }
    }
    
	@Override
	public void characters(char[] ch,int start,int length) throws SAXException{
		super.characters(ch, start, length);
		
		String nodeValue = new String(ch,start,length);
		if(!nodeValue.trim().isEmpty()){
			
			if(isQuoteRecord){
				record.put(currentTag, nodeValue);
			}
			
			if(isQuoteHead){
				switch(currentTag){
					case "datadate":datadate = nodeValue;break;
					case "orgcode":orgcode = nodeValue;break;
					case "orgname":orgname = nodeValue;break;
					case "reporter":reporter = nodeValue;break;
					case "checker":checker = nodeValue;break;
				}
			}	
		}
	}
    
    private void addData() {
    	List<String> qualifiers = new ArrayList<String>();
    	TPut put = new TPut();
		TColumnValue columnValue = null;
		List<TColumnValue> columnValues = new ArrayList<TColumnValue>();
		
		for(Map.Entry<String, String> entry : record.entrySet()){
			String key = entry.getKey();
			String value = entry.getValue();
			String qualifier = null;
			
			int index = key.indexOf("090");
			if(index > 0){
				qualifier = key.substring(0,index);
			}else{
				qualifier = key;
			}
			
			columnValue = new TColumnValue();
			columnValue.setFamily(family.getBytes());
			columnValue.setQualifier(qualifier.getBytes());
			columnValue.setValue(value.getBytes());
			
			columnValues.add(columnValue);
			qualifiers.add(qualifier);
			
    	}
		
		if(!qualifiers.contains("XMHT") && !qualifiers.contains("XMHTID"))
			isFileValid = false;
		
		//  do not hit me   i am sorry 
		columnValue = new TColumnValue();
		columnValue.setFamily(family.getBytes());
		columnValue.setQualifier("DATADATE".getBytes());
		columnValue.setValue(datadate.getBytes());
		columnValues.add(columnValue);
		
		columnValue = new TColumnValue();
		columnValue.setFamily(family.getBytes());
		columnValue.setQualifier("ORGCODE".getBytes());
		columnValue.setValue(orgcode.getBytes());
		columnValues.add(columnValue);
		
		columnValue = new TColumnValue();
		columnValue.setFamily(family.getBytes());
		columnValue.setQualifier("ORGNAME".getBytes());
		columnValue.setValue(orgname.getBytes());
		columnValues.add(columnValue);
		
		columnValue = new TColumnValue();
		columnValue.setFamily(family.getBytes());
		columnValue.setQualifier("REPORTER".getBytes());
		columnValue.setValue(reporter.getBytes());
		columnValues.add(columnValue);
		
		columnValue = new TColumnValue();
		columnValue.setFamily(family.getBytes());
		columnValue.setQualifier("CHECKER".getBytes());
		columnValue.setValue(checker.getBytes());
		columnValues.add(columnValue);
		
		put.setColumnValues(columnValues);
		byte [] row = null;
		
		switch(table){
			case "J0901":{
				indexLoan++;
				row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexLoan);
				put.setRow(row);
				putLoan.add(put);
				break;
			}
			case "J0902":{
				indexDebtor++;
				row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexDebtor);
				put.setRow(row);
				putDebtor.add(put);
				break;
			}
			case "J0903":{
				indexInvest++;
				row = RowkeyUtil.getGetRowkey(orgcode,datadate,indexInvest);
				put.setRow(row);
				putInvest.add(put);
				break;
			}
		}
	}

	public void openThrift() throws TTransportException{
    	this.transport.open();
    }
    
    public void putData(){
    	if(isFileValid){
    		try {
    			// need to improve with subList 
    			if(putLoan.size()>0){
    				long startTime = System.currentTimeMillis();
    				System.out.print(" put loan data size: "+putLoan.size());
    				if(putLoan.size()>5000){
    					List<TPut> tmp = new ArrayList<TPut>(5000);
    					for (TPut tPut : putLoan) {
							tmp.add(tPut);
							if(tmp.size()>5000){
								client.putMultiple(ByteBuffer.wrap(tableLoan.getBytes()), tmp);
								tmp.clear();
							}
						}
    					client.putMultiple(ByteBuffer.wrap(tableLoan.getBytes()), tmp);
    				}else{
    	 				client.putMultiple(ByteBuffer.wrap(tableLoan.getBytes()), putLoan);
    				}
    				long endTime = System.currentTimeMillis();
    				long putTime = endTime-startTime;
    				System.out.print(" put time: "+putTime);
    			}else{
    				System.out.print("  put loan size = 0");
    			}
			} catch (TIOError e) {
				e.printStackTrace();
			} catch (TException e) {
				e.printStackTrace();
			}
    		
    		try {
    			if(putDebtor.size()>0){
    				System.out.print(" put debtor data size: "+putDebtor.size());
    				if(putDebtor.size()>5000){
    					List<TPut> tmp = new ArrayList<TPut>(5000);
    					for (TPut tPut : putDebtor) {
    						tmp.add(tPut);
    						if(tmp.size()>5000){
    							client.putMultiple(ByteBuffer.wrap(tableDebtor.getBytes()), tmp);
    							tmp.clear();
    						}
						}
    					client.putMultiple(ByteBuffer.wrap(tableDebtor.getBytes()), tmp);
    				}else{
    					client.putMultiple(ByteBuffer.wrap(tableDebtor.getBytes()), putDebtor);
    				}
    				
    			}else{
    				System.out.print("  put debtor size = 0");
    			}
    		} catch (TIOError e) {
				e.printStackTrace();
			} catch (TException e) {
				e.printStackTrace();
			}
    		
    		try {
    			if(putInvest.size()>0){
    				System.out.println(" put invest data size: "+putInvest.size());
    				if(putInvest.size()>5000){
    					List<TPut> tmp = new ArrayList<TPut>(5000);
    					for (TPut tPut : putInvest) {
							tmp.add(tPut);
							if(tmp.size()>5000){
								client.putMultiple(ByteBuffer.wrap(tableInvest.getBytes()), tmp);
								tmp.clear();
							}
						}
    				}else{
    					client.putMultiple(ByteBuffer.wrap(tableInvest.getBytes()), putInvest);
    				}
    			}else{
    				System.out.print(" put invest size = 0");
    			}
    		} catch (TIOError e) {
				e.printStackTrace();
			} catch (TException e) {
				e.printStackTrace();
			}
    	}else{
    		System.out.println("unvalid file do not put ");
    	}
    }
    
    public void closeThrift(){
    	this.transport.close();
    }
    
    public String getDataDate(){
    	return datadate;
    }
    
}
